package com.miao.im.repository;

import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson.JSON;
import com.miao.im.common.CacheKey;
import com.miao.im.common.enums.LuaName;
import com.miao.im.domain.UserDo;
import com.miao.im.model.AuthResponseModel;
import com.miao.im.repository.dao.IUserDao;
import com.miao.im.support.RedisLuaExecutor;
import com.miao.im.support.util.JsonUtil;
import com.miao.im.support.util.ProtoBufUtil;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.ReactiveStringRedisTemplate;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Repository;
import reactor.core.publisher.Mono;

import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * @Auther: miaoguoxin
 * @Date: 2021/4/25 22:08
 * @Description:
 */
@Repository
public class UserRepository implements IUserRepository {
    @Autowired
    private RedisLuaExecutor redisLuaExecutor;
    @Autowired
    private StringRedisTemplate redisTemplate;
    @Autowired
    private ReactiveStringRedisTemplate reactiveStringRedisTemplate;
    @Autowired
    private IUserDao userDao;

    @Override
    public void updateUserInfo(UserDo userDo) {
        userDao.updateById(userDo);
        String key = CacheKey.USER_INFO_PREFIX + userDo.getId();
        redisTemplate.delete(key);
    }

    @Override
    public Optional<UserDo> getBySessionId(String sessionId) {
        String key = CacheKey.WEB_LOGIN_SESSION_PREFIX + sessionId;
        String json = redisTemplate.opsForValue().get(key);
        if (StrUtil.isBlank(json)) {
            return Optional.empty();
        }
        return Optional.ofNullable(JSON.parseObject(json, UserDo.class));
    }

    @Override
    public Mono<UserDo> reactiveGetBySessionId(String sessionId) {
        String key = CacheKey.WEB_LOGIN_SESSION_PREFIX + sessionId;
        return reactiveStringRedisTemplate.opsForValue().get(key)
                .flatMap(s -> Mono.justOrEmpty(JSON.parseObject(s, UserDo.class)));
    }

    @Override
    public Optional<UserDo> getByUserId(Long userId) {
        String key = CacheKey.USER_INFO_PREFIX + userId;
        String json = redisTemplate.opsForValue().get(key);
        if (StrUtil.isNotBlank(json)) {
            UserDo userDo = JSON.parseObject(json, UserDo.class);
            return Optional.ofNullable(userDo);
        }

        Optional<UserDo> optionalUserDo = userDao.getById(userId);
        optionalUserDo.ifPresentOrElse(
                userDo -> redisTemplate.opsForValue().set(key, JsonUtil.toJSONStr(userDo), 1, TimeUnit.HOURS),
                //防穿透
                () -> redisTemplate.opsForValue().set(key, "{}", 5, TimeUnit.MINUTES));
        return optionalUserDo;
    }

    @Override
    public Mono<Void> reactiveSaveOnlineUser(AuthResponseModel.AuthResponse userInfo) {
        String key = CacheKey.ONLINE_USER_PREFIX + userInfo.getUserId();
        return reactiveStringRedisTemplate.opsForHash().put(key, userInfo.getDeviceNumber(), "1")
                .flatMap(a -> {
                    String jsonStr = JsonUtil.toJSONStr(userInfo.toByteArray());
                    String timeoutKey = CacheKey.ONLINE_USER_TIMEOUT_PREFIX + userInfo.getUserId() + CacheKey.SEMICOLON_DELIMITER + userInfo.getDeviceNumber();
                    return reactiveStringRedisTemplate.opsForValue().set(timeoutKey, jsonStr, Duration.ofSeconds(CacheKey.ONLINE_USER_TIMEOUT_SEC));
                }).then();
    }

    @Override
    public void delOnlineUser(String userId, String deviceNumber) {
        String key = CacheKey.ONLINE_USER_PREFIX + userId;
        redisTemplate.opsForHash().delete(key, deviceNumber);

        String timeoutKey = CacheKey.ONLINE_USER_TIMEOUT_PREFIX + userId + CacheKey.SEMICOLON_DELIMITER + deviceNumber;
        redisTemplate.delete(timeoutKey);
    }

    @Override
    public void batchRefreshOnlineUser(List<String> keys) {
        List<String> collect = keys.stream().map(k -> CacheKey.ONLINE_USER_TIMEOUT_PREFIX + k).collect(Collectors.toList());
        if (CollectionUtil.isEmpty(collect)) {
            return;
        }
        Object[] args = new Object[]{
                collect.size() + "", CacheKey.ONLINE_USER_TIMEOUT_SEC + ""
        };
        redisLuaExecutor.execute(LuaName.BATCH_REFRESH_ONLINE_USER, collect, args);
    }


    @Override
    public Mono<List<AuthResponseModel.AuthResponse>> reactiveGetOnlineUsers(Long userId) {
        String key = CacheKey.ONLINE_USER_PREFIX + userId;
        return reactiveStringRedisTemplate.opsForHash().keys(key)
                .map(kO -> CacheKey.ONLINE_USER_TIMEOUT_PREFIX + userId + CacheKey.SEMICOLON_DELIMITER + kO.toString())
                .collectList()
                .flatMap(listKeys -> {
                    if (CollectionUtil.isEmpty(listKeys)) {
                        return Mono.just(new ArrayList<String>());
                    }
                    return reactiveStringRedisTemplate.opsForValue().multiGet(listKeys);
                })
                .map(jsonList -> jsonList.stream()
                        .filter(Objects::nonNull)
                        .map(this::parseAuthResponse)
                        .collect(Collectors.toList()));
    }



    private AuthResponseModel.AuthResponse parseAuthResponse(String json) {
        byte[] bytes = JSON.parseObject(json, byte[].class);
        return ProtoBufUtil.parse(bytes, AuthResponseModel.AuthResponse.class);
    }

}
